package com.kafka.producer;

import com.alibaba.fastjson.JSON;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Properties;

/**
 * @describte: kafka生产者
 * @program:foodie-dev
 * @author:zhengdp
 * @create:2021-09-18 11:32:27
 **/
public class CollectKafkaProducer {


    private final KafkaProducer<String,String> producer;

    private final String topic;

    public CollectKafkaProducer(String topic){
        // 生成Producer的配置对象
        Properties props = new Properties();
        // 配置Producer配置
        // 配置 broker地址
        props.put("bootstrap.servers","192.168.206.70:9092");
        // 配置 client.id
        props.put("client.id","demo-producer-test");
        /*
           其他的一些配置项
            props.put("batch.size", 16384);      //16KB -> 满足16KB发送批量消息
            props.put("linger.ms", 10);       //10ms -> 满足10ms时间间隔发送批量消息
            props.put("buffer.memory", 33554432);   //32M -> 缓存提性能
         */
        // 序列化配置
        props.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");

        // 创建 kafka producer 和 主题
        producer = new KafkaProducer<String, String>(props);
        this.topic = topic;
    }

    /**
     * 发送消息，根据 syncSend 来判断发送同步还是异步消息
      * @param message 消息
     * @param syncSend 同步标识
     */
    public void send(Object message, boolean syncSend){
        if(syncSend){
            // 同步发送
            // 封装要发送的消息，topic,partation等信息
            ProducerRecord msg = new ProducerRecord(topic, JSON.toJSONString(message));
            producer.send(msg);
        }else{
            // 异步发送
            ProducerRecord msg = new ProducerRecord(topic,JSON.toJSONString(message));
            // 回调方法
            Callback callback = new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if(exception!=null){
                        // 发送失败
                        System.err.println("Unable to write to Kafka in CollectKafkaProducer [" + topic + "] exception: " + exception);
                    }
                }
            };
            producer.send(msg,callback);
        }
    }

}
